package com.opdar.athena.client.agent.server.tcp;

import com.opdar.athena.client.agent.base.Configuration;
import com.opdar.athena.base.Message;
import com.opdar.athena.tcp.TcpRecive;
import com.opdar.athena.utils.MessageUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

import java.io.EOFException;

/**
 * Created by shiju on 2017/6/27.
 */
@ChannelHandler.Sharable
public class TcpDecoder extends ChannelInboundHandlerAdapter {
    private Configuration configuration = Configuration.getConfiguration();
    private byte[] project = configuration.getProject().getBytes();
    private ByteBuf byteBuf;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //项目名称(fixed) 协议版本(int) 消息类型(short int) 长度(int) 消息内容(长度)
        ByteBuf byteBuf = (ByteBuf) msg;
        if (this.byteBuf != null) {
            byteBuf = Unpooled.wrappedBuffer(this.byteBuf, byteBuf);
            this.byteBuf = null;
        }
        while (byteBuf.readableBytes() > 0) {
            byteBuf.markReaderIndex();
            //验证协议
            if (byteBuf.readableBytes() < project.length) {
                //包长度不够,reset并退出
                byteBuf.resetReaderIndex();
                byteBuf.readBytes(this.byteBuf = Unpooled.buffer(byteBuf.readableBytes()));
                break;
            }
            byte[] _project = new byte[project.length];
            byteBuf.readBytes(_project);
            if (compareProject(_project)) {
                if (byteBuf.readableBytes() >= 10) {
                    int version = byteBuf.readInt();
                    int type = byteBuf.readShort();
                    int length = byteBuf.readInt();
                    if (byteBuf.readableBytes() >= length) {
                        byte[] body = new byte[length];
                        byteBuf.readBytes(body);
                        Message message = MessageUtil.newMessageTypeInstance(type, body);
                        if (message != null) {
                            ctx.fireChannelRead(new TcpRecive(type,message));
                        }
                    } else {
                        byteBuf.resetReaderIndex();
                        byteBuf.readBytes(this.byteBuf = Unpooled.buffer(byteBuf.readableBytes()));
                        break;
                    }
                } else {
                    byteBuf.resetReaderIndex();
                    byteBuf.readBytes(this.byteBuf = Unpooled.buffer(byteBuf.readableBytes()));
                    break;
                }
            } else {
                byteBuf.resetReaderIndex();
                ReferenceCountUtil.retain(byteBuf);
                ctx.fireChannelRead(byteBuf);
            }
        }
        ReferenceCountUtil.safeRelease(byteBuf);
    }

    private boolean compareProject(byte[] _project) throws EOFException {
        boolean ret = true;
        for (int i = 0; i < project.length; i++) {
            byte b = _project[i];
            if (project[i] != b) {
                ret = false;
                break;
            }
        }
        return ret;
    }

}
